package svc

import (
	"fmt"
	"sync/atomic"
	"time"

	"gitee.com/lyuanbo/stock-plus/pkg/consts"
	"github.com/alitto/pond/v2"
	"github.com/bytedance/sonic"
	"github.com/go-redsync/redsync/v4"
	"github.com/google/uuid"
	"github.com/hibiken/asynq"
	"github.com/nats-io/nats.go/jetstream"
	"github.com/redis/go-redis/v9"
	"github.com/robfig/cron/v3"
	"go.uber.org/zap"
)

type StockSvc struct {
	svcCtx   *ServiceContext
	isLeader atomic.Bool

	leaderLock       *redsync.Mutex
	schedulerObjects map[string]string

	asynqClient *asynq.Client
}

func NewStockService(svcCtx *ServiceContext) *StockSvc {
	s := &StockSvc{
		svcCtx:           svcCtx,
		schedulerObjects: make(map[string]string),
	}

	s.leaderLock = svcCtx.RedSync.NewMutex("stock:leader",
		redsync.WithExpiry(12*time.Second),           // 4倍任务间隔
		redsync.WithTries(2),                         // 允许1次重试
		redsync.WithRetryDelay(300*time.Millisecond), // 重试延迟
		redsync.WithDriftFactor(0.01),
		redsync.WithTimeoutFactor(0.1), // 增加超时容忍度
		redsync.WithGenValueFunc(func() (string, error) {
			uid, _ := uuid.NewV7()
			return uid.String(), nil
		}),
	)

	s.asynqClient = asynq.NewClient(asynq.RedisClientOpt{Addr: s.svcCtx.Conf.Get().Redis.Addr, Password: s.svcCtx.Conf.Get().Redis.Password, DB: 11})

	go func() {
		for {
			if err := s.leaderLock.Lock(); err != nil {
				s.isLeader.Store(false)
				time.Sleep(1 * time.Second)
				continue
			}

			s.isLeader.Store(true)

			// 持有锁期间做续租，而不是立即释放
			for s.isLeader.Load() {
				time.Sleep(2 * time.Second) // 2秒续租一次

				// 续租
				ok, err := s.leaderLock.Extend()
				if !ok || err != nil {
					s.isLeader.Store(false)
					break
				}
			}

			_, err := s.leaderLock.Unlock()
			if err != nil {
				s.svcCtx.Logger.Error(s.svcCtx.Ctx, "failed to unlock leader lock", zap.Error(err))
			}
			time.Sleep(1 * time.Second)
		}
	}()

	return s
}

func (s *StockSvc) Run() {
	s.registerScheduler()
}

func (s *StockSvc) Start() {
	s.syncKline()
	go s.Run()
}

func (s *StockSvc) registerScheduler() {
	cr := cron.New(cron.WithSeconds())

	_, _ = cr.AddFunc("@every 3s", s.taskRealtime) // 实时行情
	_, _ = cr.AddFunc("0 1 15 * * *", s.syncKline) // 同步K线数据

	cr.Start()

	// 启动优先调用
	s.taskRealtime()
}

func (s *StockSvc) taskRealtime() {
	arr, err := s.svcCtx.Stock.Tencent.Realtime()
	if err != nil {
		s.svcCtx.Logger.Error(s.svcCtx.Ctx, "failed to get realtime data", zap.Error(err))
		return
	}

	// 推送
	pipe := s.svcCtx.Redis.Pipeline()
	for _, v := range arr {
		buf, _ := v.KlineInfo.MarshalArray()
		pipe.ZAdd(s.svcCtx.Ctx, consts.RedisKeyStockKline(v.StockInfo.Symbol, "realtime"), redis.Z{Score: float64(v.KlineInfo.Timestamp.Unix()), Member: buf})

		buf2, _ := sonic.Marshal(v)
		_, err := s.svcCtx.JetStream.Publish(s.svcCtx.Ctx, consts.TopicRealTime, buf2, jetstream.WithMsgID(fmt.Sprintf("%s-%d", v.StockInfo.Symbol, v.KlineInfo.Timestamp.Unix())))
		if err != nil {
			s.svcCtx.Logger.Warn(s.svcCtx.Ctx, "failed to publish realtime data", zap.Error(err))
			continue
		}
	}
	_, _ = pipe.Exec(s.svcCtx.Ctx)
}

func (s *StockSvc) syncKline() {
	symbols := s.svcCtx.Stock.XueQiu.GetSymbols()

	start := time.Now()
	pool := pond.NewPool(10)

	pipe := s.svcCtx.Redis.Pipeline()
	for _, symbol := range symbols {

		pool.Submit(func() {
			realtimeKline, _ := s.svcCtx.Redis.ZRevRangeWithScores(s.svcCtx.Ctx, consts.RedisKeyStockKline(symbol, "realtime"), 0, 0).Result()
			m1Kline, _ := s.svcCtx.Redis.ZRevRangeWithScores(s.svcCtx.Ctx, consts.RedisKeyStockKline(symbol, "1m"), 0, 0).Result()

			for _, period := range s.svcCtx.Conf.Get().CCI {
				if s.svcCtx.Redis.Exists(s.svcCtx.Ctx, consts.RedisKeyStockKline(symbol, period)).Val() == 1 {
					if len(m1Kline) > 0 && (len(realtimeKline) == 0 || m1Kline[0].Score >= realtimeKline[0].Score) {
						continue
					}
				}
				key := consts.RedisKeyStockKline(symbol, period)

				var lastTime int64
				lastZ, _ := s.svcCtx.Redis.ZRevRangeWithScores(s.svcCtx.Ctx, key, 0, 0).Result()
				if len(lastZ) > 0 {
					lastTime = int64(lastZ[0].Score)
				}

				// 同步K线数据
				m := s.svcCtx.Stock.XueQiu.Kline(symbol, []string{period})
				for _, v := range m[period] {

					if v.KlineInfo.Timestamp.Unix() < lastTime {
						continue
					}
					buf, _ := v.KlineInfo.MarshalArray()
					pipe.ZAdd(s.svcCtx.Ctx, key, redis.Z{Score: float64(v.KlineInfo.Timestamp.Unix()), Member: buf})
				}
			}
		})
	}
	pool.StopAndWait()
	_, _ = pipe.Exec(s.svcCtx.Ctx)

	s.svcCtx.Logger.Info(s.svcCtx.Ctx, "sync kline time", zap.Duration("cost", time.Since(start)))
}
